package com.zs.rocketmqstream.oldway;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationRunner;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.binder.PollableMessageSource;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;

/**
 * @author madison
 * @description
 * @date 2021/7/4 15:22
 */
//@EnableBinding(PolledProcessor.PolledProcessorInterface.class)
//@Configuration
public class PolledProcessor {
    private final Logger logger =
            LoggerFactory.getLogger(PolledProcessor.class);

    // Using reactor API
    //    @Bean
    public ApplicationRunner runner(PollableMessageSource source,
                                    MessageChannel dest) {
        return args -> {
            while (true) {
                boolean result = source.poll(m -> {
                    String payload = (String) m.getPayload();
                    logger.info("Received: " + payload);
                    dest.send(MessageBuilder.withPayload(payload.toUpperCase())
                            .copyHeaders(m.getHeaders())
                            .build());
                }, new ParameterizedTypeReference<String>() {
                });
                if (result) {
                    logger.info("Processed a message");
                } else {
                    logger.info("Nothing to do");
                }
                Thread.sleep(5_000);
            }
        };
    }

    public interface PolledProcessorInterface {
        @Input("PolledProcessorInput")
        PollableMessageSource source();

        @Output("PolledProcessorOutput")
        MessageChannel dest();
    }
}
